package hbase;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class MapredRowCount {

	static final String name = "rowcounter";
	
	static class RowCounterMapper extends TableMapper<ImmutableBytesWritable, Result>{
		
		public static enum Counters {ROWS}

		@Override
		protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
			
			List<KeyValue> values = value.list();
			for(KeyValue keyValue : values){
				
				System.out.println(new String(keyValue.getValue()));
				if(keyValue.getValue().length > 0){
					
					context.getCounter(Counters.ROWS).increment(1);
				}
			}
		}
	}
	
	public static Job createSubmitTableJob(Configuration conf, String[] args) throws IOException{
		
		String tableName = args[0];
		Job job = new Job(conf, name + "_" + tableName);
		job.setJarByClass(RowCounterMapper.class);
		
		Scan scan = new Scan();
//		scan.setFilter(new FirstKeyOnlyFilter());
		scan.setCaching(1000);
		scan.setCacheBlocks(false);
		if(args != null){
			
			for(int i = 1; i < args.length; i++){
				
				String columnName = args[i];
				String[] fields = columnName.split(":");
				if(fields.length == 1){
					
					scan.addFamily(Bytes.toBytes(fields[0]));
				}else{
					
					scan.addColumn(Bytes.toBytes(fields[0]), Bytes.toBytes(fields[1]));
				}
			}
			
			job.setOutputFormatClass(NullOutputFormat.class);
			job.setNumReduceTasks(0);
			TableMapReduceUtil.initTableMapperJob(tableName, scan, RowCounterMapper.class, 
					ImmutableBytesWritable.class, Result.class, job);
		}
		
		return job;
	} 
	
	
	//args="user teacher:chinese teacher:english teacher:math"
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		
		Configuration conf = HBaseConfiguration.create();
		conf.set("hbase.zookeeper.quorum", "master");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(otherArgs.length < 1){
			
			System.err.println("error");
			System.exit(-1);
		}
		
		Job job = createSubmitTableJob(conf, otherArgs);
		job.waitForCompletion(true);
	}
}
